package com.niit.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
 * Date:2025/5/13
 * Author：Ys
 * Description:
 */
object Spark_Base1 {

  def main(args: Array[String]): Unit = {
    //准备环境
    /*1.传统创建方式*/
//    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSQL")
//    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    /*2.简单创建方式*/
    val spark = SparkSession.builder().master("local[*]").appName("sparkSQL").getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")

    //1.DataFrame
    //1.1 使用Spark SQL 读取 json文件， 生成DataFrame
    val dataFrame: DataFrame = spark.read.json("input/user.json")
    //1.2 展示DataFrame信息
    dataFrame.show()
    //1.3想使用SQL语句进行操作，需要注册为临时表，并给该临时表起名字，后续对个临时表进行操作。
    dataFrame.createOrReplaceTempView("user")
    //1.4 使用SQL语句进行查询
    val resDF1: DataFrame = spark.sql(" select * from user ")
    resDF1.show()
    //1.5查看DataFrame结构
    dataFrame.printSchema()
    //1.6只查询 username列 数据
    val resDF2: DataFrame = spark.sql(" select username from user ")
    val resDF3: DataFrame = dataFrame.select("username")
    resDF3.show()
    //1.7查询 username列 以及 “age + 1” 的数据
    //需要使用DSL :管理结构化数据的特定领域的语言，使用DSL不用创建临时表
    import spark.implicits._
    dataFrame.select($"username", $"age" + 1 as "newAge") .show()
    dataFrame.select('username,'age + 1 as "new Age").show

    //1.8 查看 age > 30 数据
    dataFrame.filter($"age" > 30).show()
    //1.9按照 age 分组， 查看数据的条数
    dataFrame.groupBy("age").count().show()

    //1.10 RDD 转换成 DataFrame
    val rdd1: RDD[Int] = spark.sparkContext.makeRDD((1 to 4).toList)
    val df2: DataFrame = rdd1.toDF("ID")
    df2.show()
    //1.11 DataFrame  转换成 RDD
    val rdd2: RDD[Row] = df2.rdd
     rdd2.collect().foreach(println)


    //2.DataSet 是一种请类型的数据表，会与样例类配合使用
    //2.1使用样例类创建DataSet
    val list = List( Person("张三",40),Person("李四",50)  )
    val ds1: Dataset[Person] = list.toDS()
    ds1.show()
    //2.2使用基本数据类型创建DataSet
    val ds2: Dataset[Int] = List(1, 2, 3, 4, 5).toDS //基本数据类型转换成Dataset默认字段名是value
    ds2.show()

    //2.3DataSet转换成RDD
    val rdd3: RDD[Person] = ds1.rdd
    rdd3.collect().foreach(println)

    //2.4RDD转换成DataSet
    val ds3: Dataset[Person] = rdd3.toDS()
    ds3.show()
    //RDD中没有样例类  去转换 DataSet
    val rdd4: RDD[(String, Int)] = spark.sparkContext.makeRDD(List(("王五", 50), ("马六", 60)))
    //将rdd4中的数据 封装进 Person类中===> 将rdd中的元组 转换 成 Person 类
    val rdd5: RDD[Person] = rdd4.map(t => {
      Person(t._1, t._2)
    })
    val ds4: Dataset[Person] = rdd5.toDS()
    ds4.show()
    //2.5DataSet转换成DataFrame
    val df3: DataFrame = ds3.toDF()
    df3.show()

    //2.6DataFrame转换成DataSet
    val ds5: Dataset[Person] = df3.as[Person]
    ds5.show()

    spark.stop()
  }
  //创建样例类
  case class Person(name:String,age:Int)
}
